package com.lg.monitor

import java.sql.{Connection, DriverManager, PreparedStatement, Statement}

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter


class MysqlWriter extends ForeachWriter[BusInfo] {

  val driver = "com.mysql.jdbc.Driver"
  private val url = "jdbc:mysql://linux123:3306/lg_logstic?useSSL=false&characterEncoding=utf8"
  private val user = "root"
  private val pwd = "12345678"
  var connection:Connection = _
  var statement:PreparedStatement  = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    Class.forName(driver)
    connection = DriverManager.getConnection(url, user, pwd)
    true
  }

  override def process(value: BusInfo): Unit = {
    val arr: Array[String] = value.lglat.split("_")
    val sql = "INSERT INTO lg_bus_gps (deployNum, plateNum, timeStr, lng, lat) VALUES (?, ?, ?, ?, ?)"
    statement = connection.prepareStatement(sql)
    statement.setString(1, value.deployNum) //调度编号
    statement.setString(2, value.plateNum) //车牌号
    statement.setString(3, value.timeStr) //时间戳
    statement.setString(4, arr(0)) //经度
    statement.setString(5, arr(1)) //纬度
    statement.executeUpdate()
  }

  override def close(errorOrNull: Throwable): Unit = {
    connection.close()
  }
}
